package com.xrui.hbase.filter;

import com.xrui.hbase.ColumnName;
import com.xrui.hbase.DataRequest;
import com.xrui.hbase.DecodedCell;
import com.xrui.hbase.exception.IOException;
import org.apache.hadoop.hbase.filter.Filter;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.node.JsonNodeFactory;
import org.codehaus.jackson.node.ObjectNode;


/**
 * The abstract base class for filters that exclude data from Rows.
 * <p>
 * <p>Classes extending RowFilter must implement the <code>hashCode</code> and
 * <code>equals</code> methods.</p>
 */
public abstract class RowFilter {
    /**
     * The JSON node name used for the deserializer class.
     */
    private static final String DESERIALIZER_CLASS_NODE = "filterDeserializerClass";

    /**
     * The JSON node name used for the filter fields.
     */
    private static final String FILTER_NODE = "filter";

    /**
     * Deserialize a {@code RowFilter} from JSON that has been constructed
     * using {@link #toJson}.
     *
     * @param json A JSON String created by {@link #toJson}
     * @return A {@code RowFilter} represented by the JSON
     * @throws IOException in case the json cannot be read or the filter
     *                     cannot be instantiated
     */
    public static RowFilter toFilter(String json) throws java.io.IOException {
        final ObjectMapper mapper = new ObjectMapper();
        try {
            final JsonNode root = mapper.readTree(json);
            return toFilter(root);
        } catch (IOException ioe) {
            throw new IOException(ioe);
        }
    }

    /**
     * Deserialize a {@code RowFilter} from JSON that has been constructed
     * using {@link #toJson}.
     *
     * @param root A {@code JsonNode} created by {@link #toJson}
     * @return A {@code RowFilter} represented by the JSON
     * @throws IOException in case the filter cannot be instantiated
     */
    public static RowFilter toFilter(JsonNode root) {
        final String filterDeserializerClassName = root.path(DESERIALIZER_CLASS_NODE).getTextValue();
        try {
            final Class<?> filterDeserializerClass = Class.forName(filterDeserializerClassName);
            final RowFilterDeserializer filterDeserializer =
                (RowFilterDeserializer) filterDeserializerClass.newInstance();
            final RowFilter filter = filterDeserializer.createFromJson(root.path(FILTER_NODE));
            return filter;
        } catch (ClassNotFoundException cnfe) {
            throw new IOException(cnfe);
        } catch (IllegalAccessException iae) {
            throw new IOException(iae);
        } catch (InstantiationException ie) {
            throw new IOException(ie);
        }
    }

    /**
     * Describes the data the filter requires to determine whether a row should be accepted.
     *
     * @return The data request.
     */
    public abstract DataRequest getDataRequest();

    /**
     * Constructs an HBase <code>Filter</code> instance that can be used to instruct the
     * HBase region server which rows to filter.
     * <p>
     * <p>You must use the given <code>context</code> object when referencing any HBase
     * table coordinates or values.  Using a HBase row key, column family name, or column
     * qualifier name when configuring an HBase filter will result in undefined
     * behavior.</p>
     * <p>
     * <p>Here's an implementation that
     * filters out rows where the latest version of the 'info:name' is equal to 'Bob'.
     * <p>
     * <pre>
     * HBaseCell&lt;CharSequence&gt; bobCellValue = new HbaseCell&lt;CharSequence&gt;(
     *     Schema.create(Schema.Type.STRING), "Bob");
     * HBaseColumnName hbaseColumn = context.getHBaseColumnName(
     *     ColumnName.create("info", "name"));
     * SingleColumnValueFilter filter = new SingleColumnValueFilter(
     *     hbaseColumn.getFamily(),
     *     hbaseColumn.getQualifier(),
     *     CompareOp.NOT_EQUAL,
     *     context.getHBaseCellValue(bobCellValue));
     * filter.setLatestVersionOnly(true);
     * filter.setFilterIfMissing(false);
     * return new SkipFilter(filter);</pre>
     *
     * @return The HBase filter the implements the semantics of this RowFilter.
     * @throws IOException If there is an error.
     */
    public abstract Filter toHBaseFilter() throws IOException, java.io.IOException;

    /**
     * Constructs a {@code JsonNode} that describes the filter so that it may be
     * serialized.
     *
     * @return A {@code JsonNode} describing the filter
     */
    public final JsonNode toJson() {
        ObjectNode root = JsonNodeFactory.instance.objectNode();
        root.put(DESERIALIZER_CLASS_NODE, getDeserializerClass().getName());
        root.put(FILTER_NODE, toJsonNode());
        return root;
    }

    /**
     * Constructs a {@code JsonNode} that holds the data structures specific to
     * the filter.  Implementing classes should include in the return node only
     * their own fields.
     *
     * @return A {@code JsonNode} containing the filter's fields
     */
    protected abstract JsonNode toJsonNode();

    /**
     * Returns {@code Class} that is responsible for de-serializing the filter.
     * Subclasses that want to have non-default constructors with final fields can
     * define a different class to deserialize it; typically the deserializer
     * class will be a static inner class.
     *
     * @return a {@code RowFilterDeserializer} class that will be responsible
     * for de-serializing this filter
     */
    protected abstract Class<? extends RowFilterDeserializer> getDeserializerClass();

    /**
     * A helper class for converting between HBase objects and their HBase counterparts.
     */
    public abstract static class Context {
        /**
         * Converts a HBase row key into an HBase row key.
         * <p>
         * <p>This method is useful only on tables with row key hashing disabled, since hashed
         * row keys have no ordering or semantics beyond being an identifier.</p>
         *
         * @param rowKey A hbase row key.
         * @return The corresponding HBase row key.
         */
        public abstract byte[] getHBaseRowKey(String rowKey);

        /**
         * Converts a Hbase cell value into an HBase cell value.
         * <p>
         * <p>Hbase stores bytes into HBase cells by encoding an identifier for the writer schema
         * followed by the binary-encoded Avro data. Because of this, RowFilter implementations
         * that inspect the contents of a cell should use this method to obtain the bytes encoded
         * as they would be stored by Hbase.</p>
         * <p>
         * <p>For example, if you are implementing a RowFilter that only considers rows
         * where column C contains an integer value of 123, construct an HBase Filter that
         * checks whether the contents of the cell in <code>getHBaseColumnName(C)</code> equals
         * <code>getHBaseCellValue(C, DecodedCell(<i>INT</i>, 123))</code>.</p>
         *
         * @param column    Name of the column this cell belongs to.
         * @param hbaseCell A Hbase cell value.
         * @return The Hbase cell encoded as an HBase value.
         * @throws IOException If there is an error encoding the cell value.
         */
        public abstract byte[] getHBaseCellValue(ColumnName column, DecodedCell<?> hbaseCell)
            throws IOException;
    }
}
